package keter.rabbitmq.confirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

public class ConfirmCallbackSend {

  private static final Logger logger = LoggerFactory.getLogger(ConfirmCallbackSend.class);
  private final static String QUEUE_NAME = "hello_confirm";

  public static void main(String[] argv) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("192.168.1.100");
    //factory.setPort(5672);//default port
    factory.setUsername("haha");
    factory.setPassword("haha");
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    channel.queueDeclare(
            QUEUE_NAME,           //队列名称
            false,        //是否持久化
            false,       //是否排他
            false,     //不再使用时该队列是否自动删除
            null);

    channel.confirmSelect();

    //消息批量发送
    for (int i = 0; i < 100; i++) {
      sendMsg(channel, "Hello World "+i+" !");
      Thread.sleep(10);
    }

    // 消息回调：broker端批量回传给发送者ack消息，且并非以固定的批量大小回传。
    // 因此可通过比对【已发消息数】和【broker确认的ack总数】来判断成功率
    channel.addConfirmListener(
        new ConfirmListener() {
          /**
           * 处理发送失败的消息：Nacks represent messages lost by the broker
           * @param deliveryTag
           * @param multiple
           */
          @Override
          public void handleNack(long deliveryTag, boolean multiple) {
            System.out.println("nack: deliveryTag = " + deliveryTag + " multiple: " + multiple);
          }

          /**
           * 发送成功的消息：Acks represent messages handled successfully;
           * @param deliveryTag
           * @param multiple
           */
          @Override
          public void handleAck(long deliveryTag, boolean multiple) {
            System.out.println("ack: deliveryTag = " + deliveryTag + " multiple: " + multiple);
          }
        });

    System.out.println("waiting some time for callback...");
    Thread.sleep(5000);
    channel.close();
    connection.close();
  }

  private static void sendMsg(Channel channel, String message) throws IOException{
    channel.basicPublish(
            "",   //默认交换器
            QUEUE_NAME,    //路由键：使用“默认”即direct交换器时，ROUTING_KEY并没什么卵用
            null,
            message.getBytes("UTF-8")//消息
    );
    System.out.println(" [x] Sent '" + message + "'");
  }
}
